#include "basic/task_executor.h"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <queue>
#include <map>
#include "basic/log.h"
#include "basic/utils.h"

namespace basic
{
namespace task_executor
{

    class TaskExecutorInner final
    {
      public:
        TaskExecutorInner() = delete;
        TaskExecutorInner(int numberThreads)
            : m_nNumThreads(numberThreads)
        {
            if (m_nNumThreads <= 0)
            {
                throw std::exception();
            }
            for (int i = 0; i < m_nNumThreads; ++i)
            {
                m_vecThreads.emplace_back(std::thread(&TaskExecutorInner::_ThreadContext, this));
            }
        }

        ~TaskExecutorInner()
        {
            {
                std::lock_guard<std::mutex> _lk(m_cvMutex);
                m_bContinue = false;
                m_cv.notify_all();
            }
            for (auto& it : m_vecThreads)
            {
                if (it.joinable())
                {
                    it.join();
                }
            }
            m_vecThreads.clear();
            utils::ClearQueue<Task*>(m_queTask);
        }

        bool AddTask(Task* pt)
        {
            std::lock_guard<std::mutex> _lk(m_queMutex);
            m_queTask.push(pt);
            pt->m_status = ETaskStatus::READY_TO_EXECUTE;

            m_cv.notify_one();
            return true;
        }

        bool RemoveTask(Task* pt)
        {
            std::lock_guard<std::mutex> _lk(m_queMutex);
            bool ret = utils::RemoveFromQueue<Task*>(m_queTask, pt);
            if (ret)
            {
                pt->m_status = ETaskStatus::CANCELED;
            }
            return ret;
        }

      private:
        void _ThreadContext()
        {
            // M_LOG_DEBUG("resource loading thread starting! tid:", std::this_thread::get_id())
            while (true)
            {
                Task* _pt{nullptr};

                {
                    std::unique_lock<std::mutex> _lk(m_cvMutex);
                    m_cv.wait(_lk, [&]() -> bool {
                        if (m_bContinue)
                        {
                            _pt = GetTask();
                            return (bool)_pt;
                        }
                        else
                        {
                            return true;
                        }
                    });
                }

                if (!m_bContinue)
                {
                    break;
                }

                _pt->Execute();
                _pt->m_status = ETaskStatus::FINISHED_EXECUTING;
            }
            // M_LOG_DEBUG("resource loading thread terminated! tid:", std::this_thread::get_id())
        }

        Task* GetTask()
        {
            std::lock_guard<std::mutex> _lk(m_queMutex);
            if (m_queTask.empty())
            {
                return nullptr;
            }
            Task* _pt = m_queTask.front();
            _pt->m_status = ETaskStatus::EXECUTING;
            m_queTask.pop();
            return _pt;
        }

      private:
        const int m_nNumThreads{1};
        bool m_bContinue{true};
        std::vector<std::thread> m_vecThreads;
        std::condition_variable m_cv;
        std::mutex m_cvMutex;
        std::queue<Task*> m_queTask;
        std::mutex m_queMutex;
    };

    std::map<TaskExecutor*, TaskExecutorInner*> g_mapExecutor;

    TaskExecutor::TaskExecutor(int numberThreads)
    {
        auto _it = g_mapExecutor.find(this);
        if (_it == g_mapExecutor.end())
        {
            g_mapExecutor.insert({this, new TaskExecutorInner(numberThreads)});
        }
        else
        {
            M_LOG_ERROR("there is ALREADY a TaskExecutorInner instance exist!,this:", this, "\tinner:", _it->second)
        }
    }

    TaskExecutor::~TaskExecutor()
    {
        auto _it = g_mapExecutor.find(this);
        if (_it == g_mapExecutor.end())
        {
            M_LOG_ERROR("there is NO TaskExecutorInner instance exist!")
        }
        else
        {
            g_mapExecutor.erase(_it);
            M_SAFE_DELETE(_it->second);
        }
    }

    bool TaskExecutor::AddTask(Task* pt)
    {
        auto _it = g_mapExecutor.find(this);
        TaskExecutorInner* _exeInner = _it->second;
        return _exeInner->AddTask(pt);
    }

    bool TaskExecutor::RemoveTask(Task* pt)
    {
        auto _it = g_mapExecutor.find(this);
        TaskExecutorInner* _exeInner = _it->second;
        return _exeInner->RemoveTask(pt);
    }

} // namespace task_executor
} // namespace basic
